mod chat_service;
use crate::chat_service::{parse_tcpstream_to_type, EchoRequest, EchoResponse};
use log::info;
use log4rs;
use protobuf::{Message, ProtobufResult};
use protoc_rust;
use std::borrow::{Borrow, BorrowMut};
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::io::{Error, Read, Write};
use std::net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
use std::rc::{Rc, Weak};
use std::str::FromStr;
use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
use std::sync::mpsc::{channel, Sender, SyncSender};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::Duration;

const ERR_MSG_STARTED: &'static str = "该server已经启动!";
// 未启动状态
const STATE_UN_STARTED: u8 = 0;
// 启动状态
const STATE_STARTED: u8 = 1;

// tcp客户端
#[derive()]
pub struct TcpClient {
    // 维护的所有连接
    all_connect: HashMap<String, Vec<TcpStream>>,

    // 处理tcpStream的handler
    stream_handler: Option<fn(&mut TcpStream)>,
}

// 客户端的通用功能接口
pub trait Client {
    fn new() -> TcpClient;

    // addr为要连接的地址，格式： [ip:port] ，例如: 192.168.0.1:1000
    fn connect(&mut self, addr: String) -> Result<(), &str>;

    // 关闭连接, remoteAddr为远端地址， index为要关闭的连接的序号（从0开始），如果index为None,则默认关闭remoteAddr的所有连接
    fn close_conn(&mut self, remote_addr: String, index: Option<usize>) -> Result<(), &str>;

    // 查看与addr的连接数
    fn conn_online(&self, addr: String) -> usize;

    // todo: (作废) 获取stream,index为连接的标号.
    fn get_stream(&mut self, addr: &str, index: Option<usize>) -> Result<&mut TcpStream, &str>;

    // 设置处理tcpStream的handler
    fn set_stream_handler(&mut self, handler: fn(&mut TcpStream));

    // 清楚所有连接
    fn clear_all(&mut self);
}

impl Client for TcpClient {
    fn new() -> TcpClient {
        let mut allConn: HashMap<String, Vec<TcpStream>> = HashMap::new();
        TcpClient {
            all_connect: allConn,
            stream_handler: None,
        }
    }

    fn connect(&mut self, mut remoteAddr: String) -> Result<(), &str> {
        // 连接远端之前先检查处理stream的handler是否已经设置
        if self.stream_handler.is_none() {
            return Err("stream_handler not set !");
        }

        let clone = remoteAddr.clone();
        let clone2 = clone.clone();
        let rc = Rc::new(remoteAddr);
        let stream = TcpStream::connect(clone).expect("fail connect");
        let v = match self.all_connect.get_mut(&(*rc.clone())) {
            Some(t) => t,
            None => {
                self.all_connect.insert(clone2, Vec::new());
                self.all_connect.get_mut(&(*rc.clone())).unwrap()
            }
        };
        v.push(stream);
        let size = v.len();
        let stream_ref = v.get_mut(size - 1).unwrap();
        self.stream_handler.unwrap()(stream_ref);

        Ok(())
    }

    fn close_conn(&mut self, addr: String, index: Option<usize>) -> Result<(), &str> {
        match self.all_connect.get_mut(&addr) {
            None => Ok(()),
            Some(v) => {
                if index.is_some() && index.unwrap() <= (v.len() - 1) {
                    let _ = v.remove(index.unwrap());
                }
                return Ok(());
            }
        }
    }

    fn conn_online(&self, addr: String) -> usize {
        let connects = self.all_connect.get(&addr);
        if connects.is_none() {
            return 0;
        }
        connects.unwrap().len()
    }

    fn get_stream(&mut self, addr: &str, mut index: Option<usize>) -> Result<&mut TcpStream, &str> {
        if index.is_none() {
            index = Some(0)
        }

        if self.all_connect.contains_key(addr) {
            match self.all_connect.get_mut(addr) {
                None => {}
                Some(list) => {
                    if list.len() == 0 {
                        return Err("连接不存在");
                    } else {
                        return Ok(list.get_mut(index.unwrap()).unwrap());
                    }
                }
            }
        }
        Err("连接不存在")
    }

    fn set_stream_handler(&mut self, handler: fn(&mut TcpStream)) {
        self.stream_handler = Some(handler);
    }

    fn clear_all(&mut self) {
        for (k, v) in self.all_connect.iter_mut() {
            for i in v.iter_mut() {
                i.shutdown(Shutdown::Both);
            }
        }
        self.all_connect.clear();
    }
}

// tcp服务端
#[derive(Clone)]
pub struct TcpServer {
    // 服务器状态，0-未启动，1-启动
    state: Arc<AtomicU8>,
    // 端口
    port: u16,
    // 地址
    localAddr: IpAddr,
    // 处理accept的handler
    acceptHandler: Option<fn(&mut TcpStream, &SocketAddr)>,
}

pub trait Server {
    // 创建
    fn new(port: Option<u16>, localAddr: Option<&str>) -> Result<TcpServer, ()>;

    // 设置处理accept的函数
    fn setAcceptHandler(&mut self, handler: fn(&mut TcpStream, &SocketAddr));

    // 启动server
    fn start(&self) -> Result<(), &str>;

    // 关闭server
    fn close(&self) -> Result<(), &str>;
}

impl Server for TcpServer {
    // 创建一个TcpServer
    fn new(mut port: Option<u16>, mut localAddr: Option<&str>) -> Result<TcpServer, ()> {
        if port.is_none() {
            port = Some(8888u16);
        }

        if localAddr.is_none() {
            localAddr = Some(&"127.0.0.1");
        }

        Ok(TcpServer {
            state: Arc::new(AtomicU8::new(STATE_UN_STARTED)),
            port: port.unwrap(),
            localAddr: IpAddr::from_str(localAddr.unwrap()).unwrap(),
            acceptHandler: None,
        })
    }

    // 设置处理accept的函数
    fn setAcceptHandler(&mut self, handler: fn(&mut TcpStream, &SocketAddr)) {
        self.acceptHandler = Some(handler);
    }

    // 启动server
    fn start(&self) -> Result<(), &str> {
        // 检查acceptHandler是否设置
        if self.acceptHandler.is_none() {
            return Err("acceptHandler is none!");
        }

        // SeqCst是内存顺序的一种，排序一致性
        match self.state.load(Ordering::SeqCst) {
            STATE_UN_STARTED => {
                let result = TcpListener::bind(SocketAddr::new(self.localAddr, self.port));
                if result.is_err() {
                    info!("{}", result.err().unwrap());
                    return Err("启动失败");
                }

                // 修改状态为启动
                self.state.store(STATE_STARTED, Ordering::SeqCst);

                let state_copy = Arc::clone(&self.state);

                let handler = self.acceptHandler.clone();

                // 开启子线程处理链接
                thread::spawn(move || {
                    let listener = result.unwrap();
                    while state_copy.load(Ordering::SeqCst) == STATE_STARTED {
                        // 处理accept
                        match listener.accept() {
                            Ok((mut stream, socketAddr)) => {
                                // 执行handler
                                handler.unwrap()(&mut stream, &socketAddr);
                            }

                            Err(e) => (println!("occour err:{}", e)),
                        };
                    }
                });

                Ok(())
            }
            STATE_STARTED => Err(ERR_MSG_STARTED),
            _ => Err("无法处理的状态值!"),
        }
    }

    fn close(&self) -> Result<(), &str> {
        // change state from 1 to 0
        let result = self.state.compare_exchange(
            STATE_STARTED,
            STATE_UN_STARTED,
            Ordering::SeqCst,
            Ordering::SeqCst,
        );
        match result {
            Ok(_) => Ok(()),
            Err(e) => {
                println!("{}", e);
                Err("关闭TcpServer失败!")
            }
        }
    }
}

//  listen port with tcp, if port is None ,then will listen default port of 8080
pub fn listen_port_on_tcp(mut port: Option<u16>, ip: IpAddr) -> Result<TcpListener, Error> {
    match port {
        None => {
            port = Some(8080);
        }
        Some(_) => {}
    }

    let tcp_listener = TcpListener::bind(SocketAddr::new(ip, port.unwrap())).unwrap();
    println!("tcp listen started!");
    Ok(tcp_listener)
}

//  connect addr
pub fn connect_to<A: ToSocketAddrs>(addr: A) {
    let mut tcp_stream = TcpStream::connect(addr).expect("fail connect");
    let mut request = EchoRequest::new();
    let len = 1025;
    let mut buff_str = Vec::with_capacity(len);
    for i in 0..len {
        buff_str.push('a' as u8);
    }
    let msg = String::from_utf8(buff_str).unwrap();
    // request.set_message("hello server ,i am client.".to_string());
    request.set_message(msg);
    let msg_bytes = request.write_to_bytes().unwrap();
    let size = tcp_stream.write(&msg_bytes).unwrap();
    println!("connect addr finish, size:{}", size as i32);
}

pub fn echo(req: EchoRequest) -> EchoResponse {
    let msg = req.get_message();
    println!("消息内容:{}", msg);
    let mut resp = EchoResponse::new();
    resp.set_message("消息已收到".to_string());
    resp
}

// 默认的 处理accept的handler
pub fn default_acceptHandler(stream: &mut TcpStream, socket: &SocketAddr) {
    let request = EchoRequest::parse(stream);

    let resp = echo(request);
    let resp_bytes = resp.write_to_bytes().unwrap();
    //  return response
    stream.write(&resp_bytes).unwrap();
}

pub fn default_server_msg_for_client(stream: &mut TcpStream) {
    let response = EchoResponse::parse(stream);
    println!("[client]收到server端的消息：{}", response.get_message());
}

// 解析protobuf的文件为rust代码
#[test]
fn test_parse_protobuf_file() {
    protoc_rust::Codegen::new()
        .out_dir("src")
        .inputs(&["chat_service.proto"])
        // .include("/Users/mac/workspace/gitee_porject/learn-rust/socket")
        .run()
        .expect("Running protoc failed.");
}

// 测试字符串转换成byte字节数组和 使用protobuf转换成字节数组的差异
#[test]
fn test_compare_str_and_protobuf() {
    let len = 1025;
    let mut buffStr = Vec::with_capacity(len);
    for i in 0..len {
        buffStr.push('a' as u8);
    }
    // let str = "hello,  test";
    // let msg = str.to_string();
    let msg = String::from_utf8(buffStr).unwrap();
    let msg2 = msg.clone();
    let m_v = msg.into_bytes();

    let mut echoR = EchoRequest::new();
    echoR.set_message(msg2);
    let e_v = echoR.write_to_bytes().unwrap();
    println!("m_v.len :{}", m_v.len());
    println!("e_v.len :{}", e_v.len());
    // assert_eq!(m_v, e_v);
    let result_m = EchoRequest::parse_from_bytes(&m_v);
    println!("result_m ok :{}", result_m.is_ok());
    let result_e = EchoRequest::parse_from_bytes(&e_v);
    println!("result_e ok :{}", result_e.is_ok());
    println!("msg :{}", result_e.unwrap().get_message());
}
